package com.zlx.rulemk.marketing.dao;

import com.zlx.rulemk.marketing.beans.EventBean;
import com.zlx.rulemk.marketing.beans.EventCombinationCondition;
import com.zlx.rulemk.marketing.beans.EventCondition;
import com.zlx.rulemk.marketing.utils.EventIDEncodeUtil;
import com.zlx.rulemk.marketing.utils.EventUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.shaded.guava18.com.google.common.base.Stopwatch;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @Author: zhanglingxing
 * @CreateTime: 2022-07-08
 * @Description: TODO
 * @Version: 1.0
 */
@Slf4j
public class StateQuerier {

    ListState<EventBean> listState;

    public StateQuerier(ListState<EventBean> listState) {
        this.listState = listState;
    }

    /**
     * 在state中，根据组合条件及查询的时间范围，得到返回结果的1213形式字符串序列
     */
    public String getEventCombinationConditionStr(
                                EventCombinationCondition eventCombinationCondition,
                                long queryRangeStart,long queryRangeEnd) throws Exception {

        // 获取状态state中的数据迭代器
        Iterable<EventBean> eventBeans = listState.get();

        // 获取事件组合条件中的感兴趣的事件
        List<EventCondition> eventConditionList = eventCombinationCondition.getEventConditionList();

        // 迭代state中的数据
        StringBuilder sb = new StringBuilder();

        TriggerQuerier triggerQuerier = new TriggerQuerier();

        Stopwatch stopwatch = Stopwatch.createStarted();

        Long size = 0L;
        for (EventBean eventBean : eventBeans) {
            size ++;
            if (eventBean.getTimeStamp() >= queryRangeStart
                    && eventBean.getTimeStamp() <= queryRangeEnd) {
                for (int i = 1; i <= eventConditionList.size(); i++) {
                    // 判断某一个事件和原子条件是否相等 FLink里面的【E】和规则的【E】是否相等
                    EventCondition eventCondition = eventConditionList.get(i - 1);

                    if (triggerQuerier.triggerEventMatchCondition(eventBean,eventCondition)) {
                        String eventCode = EventIDEncodeUtil.getEventCode(eventBean.getEventId());
                        sb.append(eventCode);
                        break;
                    }

                }
            }
        }
        // Flink状态中的
        log.debug("规则中要求事件组合：{}，flinkState满足的字符串：{},flinkState数据大小：{}，遍历耗时：{}ms"
                ,eventCombinationCondition.getMatchPattern(),(sb.toString() == "" ? "{}" : sb.toString()),
                size,stopwatch.elapsed(TimeUnit.MILLISECONDS));

        return sb.toString();
    }

    /**
     * 在state中，根据组合条件及查询的时间范围，查询该组合出现的次数
     * 事件ID和正则匹配的次数
     * @param eventCombinationCondition
     * @return
     */
    public int queryEventCombinationConditionCount(EventCombinationCondition eventCombinationCondition,
                                                    long queryRangeStart,
                                                    long queryRangeEnd) throws Exception {

        String sequenceStr = getEventCombinationConditionStr(eventCombinationCondition,queryRangeStart,queryRangeEnd);
        int count = EventUtil.sequenceStrMatchRegexCount(sequenceStr, eventCombinationCondition.getMatchPattern());

       /* log.debug("flinkstate中查询事件组合条件，根据规则从flinkState过滤事件序列字符串：{} ,规则中正则表达式：{}, 匹配结果： {}",
                sequenceStr,eventCombinationCondition.getMatchPattern(),count);*/
        return count;
    }
}
